#include <sys/time.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <numa.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include "param.h"
#include "types.h"
#include "defs.h"

extern threadinfo tinfo[];
extern int thread_num;
extern int numa_opt;
filebuf_ring io_ring;
filebuf *fbuf_io_list [MAX_FILE_NUM];
int finished;

int monitor_thread (threadinfo *info);
int merge_thread (threadinfo *info);
int io_thread (threadinfo *info); 

int
thread_init (int mt, int knum) {
    int i;
    finished = 0;
    if (thread_num < mt+2) 
        HANDLE_ERROR ("thread_init","Too many threads.\n",0);
    thread_num = mt+2;
    tinfo[0].type = TI_T_IO;
    tinfo[1].type = TI_T_MONITOR;
    for (i = 2; i < mt+2; i++) {
        tinfo[i].type = TI_T_MERGE;
        pthread_attr_init (&(tinfo[i].attr));
        tinfo[i].in_files = 0;
        tinfo[i].out_files = 0;
        tinfo[i].arg = knum;
        tinfo[i].status = TI_S_IDLE;
        tinfo[i].numa_id = (i-2) % NUMA_NODES; 
    }
    filebuf_ring_init (&io_ring);
    return 0;
    error:
    return -1;
}

int
thread_create () {
    int ret, i;
    ret = 0;
    for (i = 0; i < thread_num; i++) {
        if (tinfo[i].type == TI_T_MONITOR)
        ret |= pthread_create (&(tinfo[i].tid), &(tinfo[i].attr),
                            (void*)monitor_thread,&(tinfo[i]));
        if (tinfo[i].type == TI_T_IO)
        ret |= pthread_create (&(tinfo[i].tid), &(tinfo[i].attr),
                            (void*)io_thread,&(tinfo[i]));
        if (tinfo[i].type == TI_T_MERGE)
        ret |= pthread_create (&(tinfo[i].tid), &(tinfo[i].attr),
                            (void*)merge_thread,&(tinfo[i]));
    }
    return ret;
}

int
thread_join () {
    int ret, i;
    ret = 0;
    void *retval;
    for (i = 0; i < thread_num; i++) {
        ret |= pthread_join (tinfo[i].tid, &retval);
    }
    return ret;
}

int
load () {
    int i, len, n, r_cnt, target;
    filebuf *tmp, *curr;
    len = 0;
    while ((tmp = filebuf_ring_get (&io_ring, FB_T_RD)) != NULL) {
        fbuf_io_list[len] = tmp;
        len++;
    }
    for (i = 0; i < len; i++) {
        if (fbuf_io_list[i]->b_in == NULL) {
            printf ("load: Unexpected ->b_in==NULL.\n");
            continue;
        }
        curr = fbuf_io_list[i];
        if (curr->status == FB_S_IFIN || curr->status == FB_S_FIN)
            continue;
        target = (curr->b_in->elem_num * sizeof (uint64));
        r_cnt = 0;
        while (r_cnt < target) {
            if ((n = read (curr->fd, curr->b_in->data,
                                target - r_cnt)) == -1) {
                printf ("Warning: Read error.\n");
                break;
                HANDLE_ERROR ("load","read",1);
            }
            r_cnt += n;
            /* EOF */
            if (n == 0) {
                curr->status = FB_S_IFIN;
                //printf ("load:EOF\n");
                close (curr->fd);
                break;
            }
        }
        curr->b_in->elem_num = r_cnt / sizeof (uint64);
        curr->b_in->curr = 0;
        if (r_cnt == 0) {
            filebuf_trans_in2un (curr);
        } else {
            filebuf_trans_in2out (curr);
        }
        //printf ("load:put %d\n", i);
        SIGNAL (curr->ex_res);
    }
    return 0;
    error:
    return -1;
}

int
store () {
    int i, len, n, w_cnt, target;
    filebuf *tmp, *curr;
    len = 0;
    while ((tmp = filebuf_ring_get (&io_ring, FB_T_WR)) != NULL) {
        fbuf_io_list[len] = tmp;
        len++;
    }
    for (i = 0; i < len; i++) {
        if (fbuf_io_list[i]->b_out == NULL) {
            printf ("store: Unexpected ->b_out==NULL.\n");
            continue;
        }
        curr = fbuf_io_list[i];
        if (curr->status == FB_S_FIN)
            continue;
        target = (curr->b_out->curr * sizeof (uint64));
        w_cnt = 0;
        while (w_cnt < target) {
            if ((n = write (curr->fd, curr->b_out->data + w_cnt, target-w_cnt)) == -1) {
                HANDLE_ERROR ("store","write",1);
            }
            w_cnt += n;
        }
        if (curr->status == FB_S_IFIN) {
            if (curr->b_out->next == curr->b_out){
                curr->status = FB_S_FIN;
                close (curr->fd);
            }
            filebuf_trans_out2un (curr);
            // printf ("store: putting to un.\n");
        } else {
            filebuf_trans_out2in (curr);
        }
        // printf ("store:put %d\n", i);
        SIGNAL (curr->ex_res);
    }
    return 0;
    error:
    return -1;
}


int
io_thread (threadinfo *info) {
    (void) info;
    printf ("IO: I/O thread started.\n");
    for (;;) {
        usleep (1);
        load ();
        store();
        if (finished && io_ring.elem_num == 0) {
            printf ("IO: I/O thread finished.\n");
            break;
        }
    }
    return 0;
}

int
merge_thread (threadinfo *info) {
    filebuf *ifbuf[MAX_KNUM];
    filebuf *ofbuf;
    filebuf_heap heap;
    int mret;
    int i, fd, fin;
    uint64 j;
    fd = file_get_out_fd ();
    fin = 0;
    printf ("MERGE: Thread %ld created.\n", info->tid);
    if (numa_opt)
        numa_run_on_node (tinfo->numa_id);
    if (fd == -1)
        HANDLE_ERROR ("merge_thread","out fd",0);
    ofbuf = filebuf_init (fd, FB_T_WR,OUTPUT_BNUM,OUTPUT_BSIZ, tinfo->numa_id);
    if (ofbuf == NULL)
        HANDLE_ERROR ("merge_thread", "ofbuf",0);
    info->out_files++;
    printf ("MERGE: Output buffer has bound with fd:%d\n",fd);
    if (ofbuf == NULL)
        HANDLE_ERROR ("merge_thread","ofbuf",0);
    printf ("MERGE: info->arg %d\n", info->arg);
    for (i = 0; i < info->arg; i++) {
        fd = file_get_in_fd ();
        printf ("MERGE: Input buffer has bound with fd:%d\n",fd);
        if (fd == -1) {
            printf ("MERGE: All input buffers loaded.\n");
            break;
        }
        info->in_files++;
        ifbuf[i] = filebuf_init (fd, FB_T_RD,INPUT_BNUM,INPUT_BSIZ, tinfo->numa_id);
        if (ifbuf[i] == NULL)
            HANDLE_ERROR ("merge_thread","ifbuf[i]",0);
        int k;
        for (k = 0; k < INPUT_BNUM; k++) {
        if (filebuf_ring_put (&io_ring, ifbuf[i]) == -1)
            HANDLE_ERROR ("merge_thread","io_ring",0);
        }
    }
    heap_init (&heap, ifbuf, i);
    for (;;) {
        if (fin) break;
        info->status = TI_S_STORING;
        WAIT (ofbuf->ex_res);
        if (ofbuf->b_in == NULL) {
            printf ("MERGE: Unexpected ofbuf->b_in == NULL.\n");
            goto error;
        }
        for (j = 0; j < ofbuf->b_in->elem_num; j++) {
            info->status = TI_S_MERGING;
            mret = heap_get_min(&heap, info ,ofbuf->b_in->data + j);
            //usleep (100000);
            //printf ("merge: curr num:%ld\n" ,*(ofbuf->b_in->data + j));
            if (mret == -1) {
                j++;
                ofbuf->status = FB_S_IFIN;
                info->status = TI_S_FINISH;
                fin = 1;
                printf ("MERGE: Thread[%ld] finished.\n", info->tid);
                break;
            }
        }
        ofbuf->b_in->curr = j;
        filebuf_trans_in2out (ofbuf);
        filebuf_ring_put (&io_ring, ofbuf);
    }
    pthread_exit ((void*)&(info->retval));
    return 0;
    error:
    info->retval = -1;
    pthread_exit ((void*)&(info->retval));
    return -1;
}

#define glue(x,y) x##y

int
monitor_thread (threadinfo *info) {
    int i, fin;
    struct timeval tvs, tve;
    (void) info;
    printf ("MONITOR: Monitor thread started.\n");
    gettimeofday (&tvs, NULL);
    for (;;) {
        sleep (1);
        fin = 1;
        //CLEAR ();
        printf ("MONITOR: Status:\n");
        #define DISPLAY(x) { \
            if (tinfo[i].status == glue(TI_S_,x))\
            printf ("Thread[M%d]:%s\t",i-2,#x);\
        }   
        printf ("Thread[I/O]:I/O ring has %d items.\n", io_ring.elem_num);
        for (i = 2; i < thread_num; i++) {
            if (tinfo[i].status != TI_S_FINISH)
                fin = 0;
            DISPLAY (IDLE);
            DISPLAY (LOADING);
            DISPLAY (MERGING);
            DISPLAY (STORING);
            DISPLAY (FINISH);
            if ((i-2) % 4 == 3) printf("\n");
        }
        printf ("\n");
        if (fin) goto finish;
    }
    finish:
    fflush (stdout);
    gettimeofday (&tve, NULL);
    printf ("MONITOR: Time use:%lds.\n", tve.tv_sec - tvs.tv_sec);
    printf ("MONITOR: All merge threads finished. Waiting for I/O...\n");
    finished = 1;
    pthread_exit (NULL);
}
